一：TAO 循环——Agent 的核心架构

TAO 循环（Think → Act → Observe）。这个循环是所有 Agent 架构的共同基础，无论你后续使用 ReAct、Plan-and-Execute 还是多 Agent 编排，底层都是 TAO 循环的变体。在理解 TAO 循环之前，我们先回顾其理论溯源——从思想链（CoT）到 ReAct 框架的演进。

1. 思想链（Chain-of-Thought）

CoT 的核心思想是：通过将复杂问题分解为多个逻辑步骤，让 LLM 按顺序推理，从而提高准确率。

CoT 的两个关键机制：

1. 分解问题：将复杂任务拆解为更小的子步骤
2. 顺序思维：每一步建立在上一步的结果之上

示例：商店价格计算

问题：一家商店以 100 元的价格出售产品。如果商店降价 20%，然后加价 10%，产品的最终价格是多少？

CoT 推理过程：

• 步骤 1：计算降价 20% 后的价格：100 × (1 - 0.2) = 80 元
• 步骤 2：计算上涨 10% 后的价格：80 × (1 + 0.1) = 88 元
• 结论：最终售价为 88 元

CoT 的局限性

虽然 CoT 显著提升了 LLM 的推理能力，但它有一个致命缺陷：==在推理的中间阶段，如果某一步出现错误，错误会沿着推理链传播，导致最终答案完全错误。==更糟糕的是，LLM 无法自我验证中间步骤的正确性——它只能"想"，不能"做"。

这就是 ReAct 要解决的问题：通过引入"行动"（Action）和"观察"（Observe）环节，让 LLM 能够在推理过程中与外部环境交互，验证中间结果，从而避免错误传播。

2. TAO 循环——ReAct 的运行机制

Think（思考）：LLM 作为"大脑"，分析当前状态和用户目标，决定下一步行动。这一步可能包括：判断任务是否完成、确定需要调用的工具、规划执行顺序、评估风险等。

Act（行动）：根据 Think 阶段的决策，调用相应的工具或生成回答。如果决定调用工具，就执行工具调用；如果判断任务已完成，就生成最终回答。

Observe（观察）：收集 Act 阶段的结果——如果是工具调用，收集工具返回的数据；如果是生成回答，观察用户的反馈。将观察结果纳入上下文，为下一轮 Think 提供输入。

TAO 循环的精妙之处在于它的自终止性——Agent 在每一轮的 Think 阶段都会判断"任务是否已经完成"，如果完成就输出最终答案并退出循环，如果未完成就继续下一轮。这意味着 Agent 可以根据任务复杂度自动调整执行步数：简单任务一轮就结束，复杂任务可能需要五轮、十轮甚至更多。

关键洞察：TAO 循环本质上是一个带反馈的控制回路。传统聊天机器人是开环系统（输入 → 输出，没有反馈），而 Agent 是闭环系统（输入 → 执行 → 观察 → 调整 → 再执行）。这就是为什么 Agent 能处理复杂任务——它可以根据中间结果动态调整策略。

3. ReAct

ReAct 的关键创新：推理跟踪（Reasoning Trace）

与 TAO 循环的关系

• Think 对应 ReAct 中的 Reasoning（推理）
• Act 对应 ReAct 中的 Acting（行动）
• Observe 对应 ReAct 中的 Observation（观察行动结果）

ReAct 如何解决 CoT 的幻觉问题

纯 CoT 的执行路径：

• Thought: 我需要知道地球的质量
• Thought: 根据我的训练数据，地球质量约为 5.972 × 10²⁴ kg
• Thought: 两倍就是 5.972 × 10²⁴ × 2 = 1.1944 × 10²⁵ kg
• Answer: 地球质量的两倍是 1.1944 × 10²⁵ kg

问题在哪？如果 LLM 在第二步"记错"了地球质量（比如记成了 6.0 × 10²⁴），后续所有计算都会基于错误数据，且 LLM 无法自我纠正。

ReAct 的执行路径：

• Thought: 我需要查询地球的准确质量
• Action: search("地球质量")
• Observation: 搜索结果显示地球质量为 5.972 × 10²⁴ kg
• Thought: 现在我需要计算两倍
• Action: calculate(5.972 e 24 * 2)
• Observation: 计算结果为 1.1944 × 10²⁵ kg
• Answer: 地球质量的两倍是 1.1944 × 10²⁵ kg

关键区别：ReAct 通过"行动 → 观察"机制，将推理过程中的关键步骤交给外部工具验证，避免了 LLM 的幻觉和计算错误。这也与论文在 HotpotQA、FEVER 等任务上的结论一致：ReAct 相比纯 CoT 和纯 Act-only 方法整体表现更优。

4. 从理论到实践：ReAct Prompt 设计模板

标准 ReAct Prompt 结构

react_prompt = """
你在一个由"思考、行动、观察、回答"组成的循环中运行。
在循环的最后，你输出一个答案。

使用"思考"来描述你对所提问题的思考。
使用"行动"来执行你可用的动作之一。
"观察"将是执行这些动作的结果。
"回答"将是分析"观察"结果后得出的答案。

你可用的动作包括：

calculate（计算）:
例如：calculate: 4 * 7 / 3
执行计算并返回数字

wikipedia（维基百科）:
例如：wikipedia: Django
返回从维基百科搜索的摘要

如果有机会，请始终在维基百科上查找信息。

示例会话：

问题：法国的首都是什么？

思考：我应该在维基百科上查找关于法国的信息
行动：wikipedia: France
PAUSE

你然后会收到：

观察：法国是一个国家。首都是巴黎。

思考：我已经找到了答案
回答：法国的首都是巴黎

现在轮到你了：
"""

Prompt 设计的三个关键要素

1. 循环机制说明：明确告诉 LLM 它处于一个循环中，需要重复"思考 → 行动 → 观察"直到任务完成
2. 工具定义：清晰描述每个工具的功能、调用格式、返回内容
3. 示例演示（Few-Shot）：通过完整示例展示期望的推理格式

LangChain 的 ReAct Prompt 变体

Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question

Begin!

Question: {input} Thought: {agent_scratchpad}

这个模板中有四个占位符：

• {tools}：工具的详细描述
• {tool_names}：工具名称列表
• {input}：用户的原始问题
• {agent_scratchpad}：保存历史推理记录

⚠️ 常见误区：很多初学者认为"只要告诉 LLM 有哪些工具就行"。实际上，示例演示（Few-Shot）是 ReAct Prompt 成功的关键——它教会 LLM"应该以什么格式输出"，而不仅仅是"应该做什么"。

二：智能体核心四要素

以下四个特征是所有 Agent 系统的共同基础。

1. 自主性 / 感知 / 推理 / 行动执行

自主性（Autonomy）——从"被指挥"到"自驱动"

自主性是 Agent 最核心的特征。一个具备自主性的 Agent，在接收到高层目标后，能够独立完成任务分解、工具选择、执行顺序规划和异常处理，而不需要人类逐步指挥。例如，当用户说"帮我调研竞品"，Agent 能自主决定：调研哪些维度、从哪些渠道获取信息、如何组织报告结构。

感知能力（Perception）——从"只读文字"到"感知世界"

传统聊天机器人的输入只有用户的文字消息。而 Agent 的感知范围要广得多——它可以通过工具获取实时数据（天气、股价、新闻）、读取文件系统中的文档、解析数据库查询结果、甚至处理图片和音频输入。更重要的是，Agent 的感知是主动的：它不是被动等待用户提供信息，而是在推理过程中主动判断"我还需要什么信息"。

推理与规划（Reasoning & Planning）——从"直觉回答"到"深思熟虑"

LLM 本身就具备一定的推理能力，但这种推理是"单次"的——给定输入，直接生成输出。Agent 的推理则是迭代式的：它可以先生成一个初步计划，执行第一步后根据结果调整后续计划，遇到障碍时回退并尝试替代方案。规划能力是 Agent 处理复杂任务的关键。

行动执行（Action Execution）——从"纸上谈兵"到"真实操作"

行动执行是 Agent 区别于所有"纯文本生成"系统的标志性能力。Agent 不仅能生成"应该怎么做"的文字描述，还能通过工具真正执行操作：发送 HTTP 请求、执行 SQL 查询、运行 Python 代码、操作文件系统、调用第三方 API。

2. Agent 四大核心特征

感知能力为 Think 阶段提供信息输入，推理与规划能力驱动 Think 阶段的决策，行动执行能力支撑 Act 阶段的工具调用，而自主性则是整个循环能够自驱运转的前提。

3. 经典架构图：Lilian Weng 的 Agent 框架

理解了四大核心特征与课程章节的映射关系之后，我们需要看一张在 Agent 领域被广泛引用的架构图，它来自 OpenAI 研究员 Lilian Weng 的经典博客文章《LLM Powered Autonomous Agents》。

📝 博客链接：https://lilianweng.github.io/posts/2023-06-23-agent/ ⚠️ 强烈建议：这篇博客是 Agent 领域的必读文献，建议课后完整阅读。

1. Planning（规划）：Agent 如何将复杂任务分解为子任务，如何制定执行计划
2. Memory（记忆）：Agent 如何存储和检索历史信息，包括短期记忆和长期记忆
3. Tool Use（工具使用）：Agent 如何调用外部工具来扩展自己的能力
4. Action（行动）：Agent 如何将决策转化为具体的执行操作

这四个模块与我们前面讲的"四大核心特征"是对应的：

理解这张架构图的价值在于：它为我们提供了一个通用的分析框架。当你在评估一个 Agent 系统时，可以从这四个维度去审视：它的规划能力如何？记忆机制是否完善？工具集是否丰富？行动执行是否可靠？

三：Agent vs Workflow 概念辨析

1. 核心区别

Workflow（工作流）是指任务的执行路径在设计时就已经确定——步骤 A 完成后执行步骤 B，步骤 B 完成后执行步骤 C，整个流程是固定的、可预测的。而 Agent 的执行路径是动态的——它在每一步都根据当前状态自主决定下一步做什么，路径在运行时才确定。

用一个具体例子来感受两者的差异。假设我们要构建一个"每日新闻摘要"系统：

Workflow 方案：每天早上 8 点 → 抓取 RSS 源 → 过滤关键词 → 调用 LLM 生成摘要 → 发送邮件。这个流程每天执行完全相同的步骤，不需要任何动态决策。

Agent 方案：用户说"帮我整理今天 AI 领域的重要新闻"→ Agent 自主决定搜索哪些来源 → 判断哪些内容值得纳入 → 决定摘要的详细程度 → 必要时追加搜索补充信息。这个流程每次执行路径都可能不同。

显然，第一个场景用 Workflow 更合适——路径固定、成本低、可靠性高；第二个场景才需要 Agent——路径不确定、需要动态判断。

2. 选型决策树

问题一：任务的执行路径是否在设计时就能完全确定？

如果你能在写代码之前就画出完整的流程图（每个步骤、每个分支都确定），那就用 Workflow。如果流程图上有"视情况而定"的节点，就需要考虑 Agent。

问题二：后续步骤的选择是否依赖前面步骤的结果？

如果"步骤 B 做什么"取决于"步骤 A 返回了什么"，且这种依赖关系在设计时无法穷举，就需要 Agent 的动态决策能力。

问题三：任务是否需要在执行过程中动态调整计划？

如果任务执行到一半发现原计划不可行，需要 Agent 自主切换策略，那就必须用 Agent。

3. 典型场景对比表

从表格中可以看出一个规律：Workflow 适合"已知路径"的自动化，Agent 适合"未知路径"的智能决策。在实际项目中，最常见的架构是"Workflow 作为骨架，Agent 作为关键节点"——用 Workflow 控制整体流程，在需要动态决策的节点嵌入 Agent。这种混合架构兼顾了可靠性和灵活性。

⚠️ 常见误区：很多初学者在学了 Agent 之后，倾向于"什么都用 Agent"。这会导致系统不可预测、调试困难、成本失控。记住：Agent 是解决"不确定性"的工具，如果任务本身是确定的，Workflow 永远是更好的选择。

四：Function Calling 底层原理与完整生命周期

1. 核心误解纠正：LLM 并不执行函数

核心误解：很多人以为 Function Calling 是"LLM 自己执行了函数"。正确理解：LLM 只生成"调用指令"（JSON 格式），代码负责真正执行

Function Calling 的完整流程分为六个步骤：

2. 真实代码逻辑拆解

第一阶段：准备本地工具和路由表

import json
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv(override=True)

# 使用Deepseek的API来调用大模型
client = OpenAI(
    api_key=os.environ.get("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)
# 1. 这是你本地真正能干活的函数（大模型并不知道它的具体实现代码）
def get_weather(location: str):
    print(f"🔧 [本地执行中] 正在查询 {location} 的天气...")

    # 这里可以是发HTTP请求、查数据库等真实操作
    if location == "北京":
        return '{"temp": 25, "condition": "晴"}'
    return '{"temp": 20, "condition": "未知"}'

# 2. 【关键抽象】建立“字符串名字”到“内存里的真实函数”的映射字典
available_functions = {
    "get_weather": get_weather,             # 这里是将字符串 "get_weather" 映射到你本地的 get_weather 函数
    # 如果有别的工具："search_database": search_database
}

# 3. 告诉大模型你有这个工具（只给说明书，不给代码）
tools_description = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "获取指定城市的天气",
        "parameters": {
            "type": "object",
            "properties": {"location": {"type": "string"}},
            "required": ["location"]
        }
    }
}]

第二阶段：第一次请求大模型

messages = [{"role": "user", "content": "北京今天热吗？"}]

# 大模型看到你的问题和工具说明书，它决定调用工具
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    tools=tools_description    # 告诉大模型你有哪些工具
)

response_message = response.choices[0].message
# 查看大模型是否调用了工具
print(response_message.tool_calls)

[ChatCompletionMessageFunctionToolCall(id='call_00_m7jxkpgw1cjHQdV29iZ6Anbl', function=Function(arguments='{"location": "北京"}', name='get_weather'), type='function', index=0)]

第三阶段：【核心】你的本地代码接管并执行 此时，response_message 里虽然没有回答，但带有 tool_calls。 你必须写代码来拦截并处理它：

# 检查大模型是不是发出了调用工具的请求
if response_message.tool_calls:
    
    # 记得把大模型的"请求调用"这条记录也放进历史对话里
    messages.append(response_message)
    
    # 遍历大模型想要调用的所有函数（有时候它会并行调用多个）
    for tool_call in response_message.tool_calls:
        
        # 1. 提取大模型建议的指令
        function_name = tool_call.function.name # 比如提取到 "get_weather"
        function_args_json = tool_call.function.arguments # 比如提取到 "{\"location\": \"北京\"}"
        
        # 2. 将大模型生成的 JSON 字符串解析为真正的 Python 字典
        function_args = json.loads(function_args_json)
        
        # 3. 【真正执行的魔法在此】
        # 通过大模型给的字符串名字，从你的映射字典里找到真正的 Python 函数内存地址
        function_to_call = available_functions.get(function_name)
        
        if function_to_call:
            # 4. 在你的本地机器上，真正执行这个函数，并传入解析好的参数！
            function_result = function_to_call(**function_args)
            print(f"✅ [本地执行完毕] 得到结果: {function_result}")
        else:
            function_result = "Error: 找不到该函数"
            
        # 5. 将执行得到的结果，打包成特定格式（role="tool"），准备发回给大模型
        messages.append({
            "tool_call_id": tool_call.id, # 必须带上这个ID，告诉大模型这是对刚才它请求的回复
            "role": "tool",
            "name": function_name,
            "content": function_result,   # 把真实结果（如 '{"temp": 25}'）塞进去
        })
# 打印最终的messages
print(messages)

🔧 [本地执行中] 正在查询 北京 的天气...
✅ [本地执行完毕] 得到结果: {"temp": 25, "condition": "晴"}
[{'role': 'user', 'content': '北京今天热吗？'}, ChatCompletionMessage(content='我来帮您查询一下北京今天的天气情况。', refusal=None, role='assistant', annotations=None, audio=None, function_call=None, tool_calls=[ChatCompletionMessageFunctionToolCall(id='call_00_m7jxkpgw1cjHQdV29iZ6Anbl', function=Function(arguments='{"location": "北京"}', name='get_weather'), type='function', index=0)]), {'tool_call_id': 'call_00_m7jxkpgw1cjHQdV29iZ6Anbl', 'role': 'tool', 'name': 'get_weather', 'content': '{"temp": 25, "condition": "晴"}'}]

第四阶段：第二次请求大模型（带着结果）

# 现在 messages 里面包含了：用户问题 -> 模型的调用请求 -> 你本地执行的结果
# 再次发给大模型，它就能看着真实结果，总结出人话了
second_response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
)

print("\n🤖 最终回答：", second_response.choices[0].message.content)

🤖 最终回答： 根据查询结果，北京今天天气**晴**，气温**25°C**。

从体感上来说：
- **25°C** 属于比较舒适的温度，不冷也不热
- 晴天意味着阳光充足，中午时段可能会感觉比较温暖
- 早晚温差可能较大，建议根据具体时段和活动安排穿衣

总体来说，今天北京天气不错，是个适合户外活动的好天气！如果您需要更详细的天气预报（比如湿度、风速等），我可以为您进一步查询。

3. 封装完整的 Function Calling 管线

def function_calling_pipeline(
    user_message: str,
    tools: list,
    tool_registry: dict,
    system_prompt: str = "你是一个有用的助手。",
    model: str = "deepseek-chat",
    verbose: bool = True
) -> str:
    """
    完整的 Function Calling 管线（单轮工具调用）

    参数：
        user_message: 用户输入
        tools: 工具定义列表（JSON Schema）
        tool_registry: 工具名称到函数的映射字典
        system_prompt: 系统提示词
        model: 模型名称
        verbose: 是否打印中间过程

    返回：
        LLM 的最终回答文本
    """
    # 步骤 1：构建消息历史
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_message}
    ]

    # 步骤 2-3：发送请求，获取 LLM 决策
    response = client.chat.completions.create(
        model=model, 
        messages=messages,
        tools=tools, 
        tool_choice="auto",
        temperature=0.7, 
        max_tokens=2048
    )
    assistant_message = response.choices[0].message

    # 如果 LLM 不需要调用工具，直接返回回答
    if not assistant_message.tool_calls:
        if verbose:
            print("💬 LLM 直接回答（未调用工具）")
        return assistant_message.content

    # 步骤 4：执行工具调用
    if verbose:
        print(f"🔧 LLM 决定调用 {len(assistant_message.tool_calls)} 个工具")

    messages.append(assistant_message.model_dump())

    for tool_call in assistant_message.tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)

        if verbose:
            print(f"  → {func_name}({func_args})")

        # 执行工具
        if func_name in tool_registry:
            result = tool_registry[func_name](**func_args)
        else:
            result = json.dumps({"error": f"未知工具：{func_name}"})

        if verbose:
            print(f"  ← {result[:200]}")  # 截断过长的输出

        # 步骤 5：将结果追加到消息历史
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": result
        })

    # 步骤 6：再次调用 LLM 生成最终回答
    final_response = client.chat.completions.create(
        model=model, 
        messages=messages,
        tools=tools, 
        temperature=0.7, 
        max_tokens=2048
    )

    final_answer = final_response.choices[0].message.content
    if verbose:
        print(f"✅ 最终回答生成完成")

    return final_answer

五：API 基础概念与工具定义规范

1. 工具定义三要素：name、description、parameters

Function Calling 的第一步，是告诉 LLM"你有哪些工具可以用"。这通过一个标准化的 JSON Schema 来实现。每个工具的定义包含三个核心字段：name（工具名称）、description（工具描述）、parameters（参数定义）。LLM 完全依赖这三个字段来决定何时调用哪个工具、传什么参数。

# 一个完整的工具定义示例：天气查询工具
weather_tool = {
    "type": "function",
    "function": {
        "name": "get_weather",                          # 工具名称，建议：小写 + 下划线 + 动词开头
        "description": (                                 # 工具描述，决定调用命运：做什么、何时调用、边界在哪里。
            "获取指定城市的当前天气信息，包括气温、天气状况和湿度。"
            "当用户询问某个城市的天气、气温、是否需要带伞等问题时，调用此工具。"
        ),
        "parameters": {                                  # 参数定义
            "type": "object",                            # 参数类型：object 表示 JSON 对象
            "properties": {
                "city": {                                # 属性名：city
                    "type": "string",                    # 基础类型：string、number、boolean
                    "description": "要查询天气的城市名称，例如：北京、上海、广州"   # 参数说明
                }
            },
            "required": ["city"]                         # 必填参数列表
        }
    }
}

1.1. name：工具的唯一标识

name 是工具的唯一标识符，LLM 在决定调用工具时会返回这个名称。命名规则很简单：使用小写字母和下划线，清晰表达工具的功能。

1.2. description：决定工具命运的关键字段

description 是整个工具定义中最重要的字段——它直接决定了 LLM 是否会在正确的时机调用这个工具。很多 Agent 的工具调用失败，根源不在代码逻辑，而在于工具描述写得不够好。

一个好的工具描述需要回答三个问题：

1. 这个工具做什么？（功能说明）
2. 什么时候应该调用它？（触发条件）
3. 它不能做什么？（能力边界，可选但推荐）

# 黄金模板示例
description = (
    "获取指定城市的当前天气信息，包括气温（摄氏度）、天气状况和湿度。"   # 功能说明
    "当用户询问某个城市的天气、气温、是否需要带伞/穿外套等问题时，"     # 触发条件
    "城市名称应为中文全称，例如：北京、上海、广州。"                    # 输入格式
    "目前仅支持中国大陆主要城市，不支持历史天气和天气预报查询。"         # 能力边界
)

1.3. parameters：参数的 JSON Schema 定义

使用标准化格式的好处是：LLM 在训练时已经见过大量 JSON Schema 样本，因此能够精准理解参数定义的含义。LLM 会根据这个定义，从用户的自然语言输入中提取出结构化的参数值。

# 一个更复杂的参数定义示例：搜索工具
search_tool_params = {
    "type": "object",                   # 参数容器类型，通常为 object
    "properties": {                     # 具体参数定义集合
        "query": {                      # 参数名：query（搜索词）
            "type": "string",           # 参数数据类型：字符串
            "description": "搜索关键词，应该是简洁明确的搜索查询" # 参数功能描述，供模型理解何时使用
        },
        "max_results": {                # 参数名：max_results（结果数）
            "type": "integer",          # 参数数据类型：整数
            "description": "返回的最大结果数量，默认为 5", # 参数功能描述
            "default": 5                # 默认值设定：若模型未提供则使用此值
        },
        "language": {                   # 参数名：language（语言）
            "type": "string",           # 参数数据类型：字符串
            "description": "搜索结果的语言偏好", # 参数功能描述
            "enum": ["zh", "en"],       # 枚举约束：限定模型只能从指定列表中选择
            "default": "zh"             # 默认值设定
        }
    },
    "required": ["query"]               # 只有 query 是必填的
}

这段参数定义展示了几个关键特性：type 指定参数类型（string、integer、boolean 等），description 帮助 LLM 理解参数含义，enum 限定可选值范围，required 标注必填参数，default 提供默认值。LLM 会根据这些信息，从用户的自然语言中精确提取参数。

六：大模型内置提示词模板与工具调用响应模式

1. tool_choice 四种模式：控制 LLM 的工具调用行为

在前面的流程中，我们一直假设 LLM 自主决定是否调用工具。但实际上，API 提供了精确控制这一行为的参数—— tool_choice

在实际项目中，我们经常需要精确控制 LLM 的工具调用行为。例如：在日志记录场景下，我们希望每次对话都强制调用日志工具；在纯文本生成场景下，我们希望禁止调用任何工具。tool_choice 参数就是为了满足这些需求而设计的。

tool_choice 四种模式行为对比

2. LLM 返回的 tool_calls 结构解析

当 LLM 决定调用工具时，它会返回一个包含 tool_calls 字段的消息。理解这个结构对正确处理工具调用至关重要。

# 获取带工具调用的响应
response = client.chat.completions.create(
    model=MODEL,
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=tools,
    tool_choice="auto"
)

assistant_message = response.choices[0].message

# 检查是否包含工具调用
if assistant_message.tool_calls:
    for tool_call in assistant_message.tool_calls:
        print(f"调用 ID：{tool_call.id}")                    # 唯一标识符
        print(f"工具类型：{tool_call.type}")                 # 通常是 "function"
        print(f"函数名称：{tool_call.function.name}")        # 要调用的函数名
        print(f"函数参数：{tool_call.function.arguments}")   # JSON 格式的参数
        print("-" * 40)

tool_call 对象字段说明

实践建议：在绝大多数场景下，"auto" 是最佳选择。"required" 适合"必须执行某个操作"的场景（如每次对话都记录日志）。"none" 适合"需要纯文本输出"的场景（如生成报告时不希望 LLM 中途调用工具）。指定工具模式适合"强制执行特定操作"的场景（如强制用户身份验证）。

七：Function Calling 故障教学与排障路径

需要先区分两个容易混淆的概念：工具调用失败（LLM 没有调用工具或调用了错误的工具）和工具执行失败（LLM 正确调用了工具，但工具执行过程中出错）。前者通常是工具定义问题，后者通常是代码实现问题。明确区分这两者，能帮助你快速定位问题所在层级。

Function Calling 统一排障顺序

这个排障顺序遵循"从外到内"的原则：先检查 LLM 能看到的信息（工具定义），再检查代码的映射关系（工具注册），然后检查消息格式（tool_call_id），最后检查执行逻辑（异常处理）。按照这个顺序排查，能够快速缩小问题范围，避免在错误的方向上浪费时间。

1. 故障类型一：参数提取错误

这是最常见的故障类型之一。表现为：LLM 决定调用工具，但返回的参数缺失、类型不匹配或格式错误，导致工具执行失败。很多初学者会认为这是 LLM 的问题，但实际上，90% 的参数提取错误都是因为工具定义中的参数描述不够清晰。

1.1. 问题复现：参数描述不清导致提取失败

import os
import json
from dotenv import load_dotenv
from openai import OpenAI

# 加载环境变量
load_dotenv()

client = OpenAI(
    api_key=os.environ.get("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)
MODEL = "deepseek-chat"

# ❌ 错误示例：参数描述过于简略
bad_weather_tool = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "获取天气",  # ← 描述过于简略
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市"  # ← 参数描述不清晰
                }
            },
            "required": ["city"]
        }
    }
}

# 更复杂的测试问题
complex_question = "我下周要去北京出差，帮我查一下那边的天气"

response = client.chat.completions.create(
    model=MODEL,
    messages=[{"role": "user", "content": complex_question}],
    tools=[bad_weather_tool],
    tool_choice="auto"
)

assistant_message = response.choices[0].message
if assistant_message.tool_calls:
    tool_call = assistant_message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    print(f"📦 提取的参数：{args}")
    # 可能出现的问题：LLM 提取了 {"city": "北京出差"} 或 {"city": "那边"}
else:
    print("💬 LLM 未调用工具")

在这个更复杂的问题中，由于参数描述不清晰，LLM 可能会提取错误的城市名称（如"北京出差"、"那边"），或者干脆不调用工具。

1.2. 问题分析：为什么参数描述如此重要

LLM 在提取参数时，完全依赖 parameters 中的 description 字段来理解"应该提取什么信息"。如果描述只写"城市"，LLM 不知道：

• 应该提取完整的城市名称还是简称？
• 应该提取"北京"还是"北京市"？
• 遇到"那边"、"这里"等代词时应该如何处理？

参数描述的核心原则是：给 LLM 提供足够的上下文和示例，让它能够从自然语言中精确提取出结构化参数。

让我们深入理解这个原理：LLM 在提取参数时，需要将自然语言映射到结构化字段。如果描述只写"城市"，LLM 需要自行推断：1）应该提取完整名称还是简称？2）遇到代词如何处理？3）格式要求是什么？描述越详细，LLM 的推断空间越小，提取准确率越高。

1.3. 修复方案：优化参数描述

# ✅ 修复版本：清晰的参数描述
good_weather_tool = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": (
            "获取指定城市的当前天气信息，包括气温（摄氏度）、天气状况和湿度。"
            "当用户询问某个城市的天气、气温、是否需要带伞/穿外套等问题时，调用此工具。"
            "目前支持的城市：北京、上海、广州、深圳、杭州。"
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": (
                        "要查询天气的城市名称，必须是完整的中文城市名（不含'市'字）。"
                        "例如：北京、上海、广州。"
                        "如果用户使用代词（如'那边'、'这里'），需要根据上下文推断具体城市名。"
                    )
                }
            },
            "required": ["city"]
        }
    }
}

# 用相同的复杂问题测试修复后的工具
response = client.chat.completions.create(
    model=MODEL,
    messages=[{"role": "user", "content": "我下周要去北京出差，帮我查一下那边的天气"}],
    tools=[good_weather_tool],
    tool_choice="auto"
)

assistant_message = response.choices[0].message
if assistant_message.tool_calls:
    tool_call = assistant_message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    print(f"✅ 修复后提取的参数：{args}")
    # 预期输出：{"city": "北京"}

1.4. 效果验证与对比

参数描述质量对比：错误版本 vs 修复版本

通过这个对比可以看出，参数描述的投入产出比极高——多写 50 个字，就能显著提升参数提取的准确性和可靠性。

1.5. 补充方案：添加参数校验

除了优化参数描述，我们还可以在工具函数内部添加参数校验，作为第二道防线：

def get_weather_safe(city: str = None) -> str:
    """带参数校验的天气查询工具"""
    # 参数校验
    if not city or not isinstance(city, str):
        return json.dumps({
            "error": "参数错误：city 必须是非空字符串",
            "received": city,
            "hint": "请提供有效的城市名称，例如：北京、上海、广州"
        }, ensure_ascii=False)

    # 支持的城市列表
    supported_cities = ["北京", "上海", "广州", "深圳", "杭州"]
    if city not in supported_cities:
        return json.dumps({
            "error": f"暂不支持查询 {city} 的天气",
            "supported_cities": supported_cities
        }, ensure_ascii=False)

    # 模拟天气数据
    weather_db = {
        "北京": {"temperature": 33, "condition": "晴", "humidity": 45},
        "上海": {"temperature": 28, "condition": "多云", "humidity": 72},
        "广州": {"temperature": 35, "condition": "雷阵雨", "humidity": 85},
        "深圳": {"temperature": 34, "condition": "晴转多云", "humidity": 78},
        "杭州": {"temperature": 30, "condition": "阴", "humidity": 68},
    }

    data = weather_db[city]
    return json.dumps({
        "city": city,
        "temperature": data["temperature"],
        "condition": data["condition"],
        "humidity": data["humidity"],
        "unit": "摄氏度"
    }, ensure_ascii=False)

# 测试参数校验
print("测试1：正常参数")
print(get_weather_safe("北京"))

print("\n测试2：空参数")
print(get_weather_safe(None))

print("\n测试3：不支持的城市")
print(get_weather_safe("纽约"))

🔥 踩坑预警：参数校验返回的错误信息会被传回 LLM，LLM 会基于错误信息生成友好的回答。因此，错误信息应该是结构化的 JSON 格式，而不是直接抛出异常。

2. 故障类型二：工具注册错误

这是一个看似低级但极其高频的错误。表现为：LLM 决定调用某个工具，但代码执行时报 KeyError，提示找不到对应的函数。根因往往是工具定义中的 name 与注册表中的函数名不一致——通常是拼写错误或大小写不匹配。

2.1. 问题复现：名称拼写错误导致工具找不到

# 定义工具（正确的名称）
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",  # ← 正确的名称
            "description": (
                "获取指定城市的当前天气信息，包括气温（摄氏度）、天气状况和湿度。"
                "当用户询问某个城市的天气、气温、是否需要带伞/穿外套等问题时，调用此工具。"
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "要查询天气的城市名称，例如：北京、上海、广州"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

# ❌ 错误示例：注册表中的名称拼写错误
def get_weather(city: str) -> str:
    """天气查询工具"""
    weather_db = {
        "北京": {"temperature": 33, "condition": "晴", "humidity": 45},
        "上海": {"temperature": 28, "condition": "多云", "humidity": 72},
    }
    data = weather_db.get(city, {"temperature": 25, "condition": "未知", "humidity": 50})
    return json.dumps({"city": city, **data}, ensure_ascii=False)

# 注册表中的名称拼写错误（少了一个 't'）
BUGGY_TOOL_REGISTRY = {
    "get_waether": get_weather,  # ← 拼写错误！应该是 get_weather
}

# 测试调用
response = client.chat.completions.create(
    model=MODEL,
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=tools,
    tool_choice="auto"
)

assistant_message = response.choices[0].message

if assistant_message.tool_calls:
    tool_call = assistant_message.tool_calls[0]
    func_name = tool_call.function.name
    func_args = json.loads(tool_call.function.arguments)

    print(f"🔧 LLM 决定调用工具：{func_name}")
    print(f"📦 参数：{func_args}")

    # 尝试执行工具（会报错）
    try:
        func = BUGGY_TOOL_REGISTRY[func_name]  # ← 这里会抛出 KeyError
        result = func(**func_args)
        print(f"✅ 执行成功：{result}")
    except KeyError as e:
        print(f"❌ 执行失败：KeyError: {e}")
        print(f"💡 原因：注册表中没有名为 '{func_name}' 的工具")
        print(f"💡 注册表中的工具：{list(BUGGY_TOOL_REGISTRY.keys())}")

🔧 LLM 决定调用工具：get_weather
📦 参数：{'city': '北京'}
❌ 执行失败：KeyError: 'get_weather'
💡 原因：注册表中没有名为 'get_weather' 的工具
💡 注册表中的工具：['get_waether']

这个错误非常隐蔽——工具定义和注册表都在代码中，但因为拼写错误，两者无法匹配。在真实项目中，如果工具数量很多，这种错误很难通过肉眼发现。

2.2 问题分析：为什么会出现名称不匹配

名称不匹配的根本原因是硬编码字符串。工具定义中写了一次 "get_weather"，注册表中又写了一次 "get_waether"，两处的字符串没有任何关联，编译器无法检查拼写错误。

核心原则：任何需要在多处使用的标识符，都应该用常量定义，而不是硬编码字符串。

2.3. 修复方案一：使用常量定义工具名

# ✅ 修复方案一：使用常量定义工具名
TOOL_NAME_WEATHER = "get_weather"

# 工具定义中使用常量
tools_fixed = [
    {
        "type": "function",
        "function": {
            "name": TOOL_NAME_WEATHER,  # ← 使用常量
            "description": (
                "获取指定城市的当前天气信息，包括气温（摄氏度）、天气状况和湿度。"
                "当用户询问某个城市的天气、气温、是否需要带伞/穿外套等问题时，调用此工具。"
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "要查询天气的城市名称，例如：北京、上海、广州"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

# 注册表中使用常量
TOOL_REGISTRY_FIXED = {
    TOOL_NAME_WEATHER: get_weather,  # ← 使用常量
}

# 测试修复后的版本
response = client.chat.completions.create(
    model=MODEL,
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=tools_fixed,
    tool_choice="auto"
)

assistant_message = response.choices[0].message

if assistant_message.tool_calls:
    tool_call = assistant_message.tool_calls[0]
    func_name = tool_call.function.name
    func_args = json.loads(tool_call.function.arguments)

    print(f"🔧 LLM 决定调用工具：{func_name}")
    print(f"📦 参数：{func_args}")

    # 执行工具
    func = TOOL_REGISTRY_FIXED[func_name]
    result = func(**func_args)
    print(f"✅ 执行成功：{result}")

2.4. 修复方案二：自动生成注册表

# ✅ 修复方案二：自动从工具定义生成注册表
def build_tool_registry(tools: list, func_map: dict) -> dict:
    """
    根据工具定义自动构建注册表，确保名称一致

    参数：
        tools: 工具定义列表（JSON Schema）
        func_map: 函数名到函数对象的映射

    返回：
        工具注册表（工具名 -> 函数对象）
    """
    registry = {}
    for tool in tools:
        name = tool["function"]["name"]
        if name in func_map:
            registry[name] = func_map[name]
        else:
            raise ValueError(f"工具 '{name}' 没有对应的函数实现，请检查 func_map")
    return registry

# 函数映射（函数名 -> 函数对象）
FUNC_MAP = {
    "get_weather": get_weather,
}

# 自动生成注册表
TOOL_REGISTRY_AUTO = build_tool_registry(tools_fixed, FUNC_MAP)

print(f"✅ 自动生成的注册表：{list(TOOL_REGISTRY_AUTO.keys())}") # ✅ 自动生成的注册表：['get_weather']

2.5. 效果验证与对比

工具注册方式对比：错误版本 vs 修复版本

💡 实践建议：如果你的项目只有 2-3 个工具，使用常量定义即可；如果工具数量超过 5 个，强烈建议使用自动生成注册表的方式，避免维护成本随工具数量线性增长。

3. 故障类型三：消息拼接错误

这是最隐蔽、最难排查的错误类型。表现为：工具执行成功了，但 LLM 的最终回答与工具结果完全不相关，或者说"我无法获取该信息"。很多初学者会怀疑是 LLM 的问题，但实际上，这通常是因为 tool_call_id 不匹配，导致 LLM 无法将工具结果与调用请求关联起来。

3.1. 问题复现：使用错误的 tool_call_id

# 注意：本节代码依赖 之前定义的 tools_fixed 和 TOOL_REGISTRY_FIXED
# 如果你是单独运行本节代码，请先运行上面的代码

# 步骤一：LLM 返回工具调用
messages = [
    {"role": "system", "content": "你是一个有用的助手，可以查询天气。"},
    {"role": "user", "content": "北京今天天气怎么样？"}
]

response = client.chat.completions.create(
    model=MODEL,
    messages=messages,
    tools=tools_fixed,
    tool_choice="auto"
)

assistant_message = response.choices[0].message
tool_call = assistant_message.tool_calls[0]

print(f"🔧 LLM 决定调用工具：{tool_call.function.name}")
print(f"🆔 LLM 返回的 tool_call_id：{tool_call.id}")

# 步骤二：执行工具
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
result = TOOL_REGISTRY_FIXED[func_name](**func_args)

print(f"📋 工具执行结果：{result}")

# 步骤三：❌ 错误的消息拼接（使用自定义 ID）
messages.append(assistant_message.model_dump())
messages.append({
    "role": "tool",
    "tool_call_id": "my_custom_id_12345",  # ← 错误：不是 LLM 返回的 ID
    "content": result
})

print(f"\n❌ 错误版本：使用自定义 ID 'my_custom_id_12345'")

# 步骤四：再次调用 LLM
final_response = client.chat.completions.create(
    model=MODEL,
    messages=messages,
    tools=tools_fixed
)

print(f"🤖 LLM 回答：")
print(final_response.choices[0].message.content)

直接抛出 400 网络错误（报错退出代码）：像代码中 client.chat.completions.create(...) 这一行发起的第二次请求，甚至都没有真正送到大模型（服务器推理引擎）的大脑里去，就直接被 API 服务器拒绝了。这就是 tool_call_id 不匹配导致的问题。

3.2. 问题分析：为什么 tool_call_id 必须精确匹配

在 Function Calling 的消息流中，LLM 需要将工具的执行结果与之前的调用请求关联起来。这个关联是通过 tool_call_id 实现的：

1. LLM 在返回工具调用时，会为每个调用生成一个唯一的 id（例如 call_abc123xyz）
2. 代码执行工具后，必须在 tool 消息中使用相同的 tool_call_id
3. LLM 收到 tool 消息后，会根据 tool_call_id 找到对应的调用请求，将结果与请求关联
4. 如果 tool_call_id 不匹配，LLM 会认为"这个工具结果不是我要的"，从而忽略它

核心原则：tool_call_id 必须与 LLM 返回的 tool_call.id 完全一致，不能使用自定义 ID，也不能省略。

3.3. 修复方案：严格使用 LLM 返回的 ID

# ✅ 修复版本：使用正确的 tool_call_id
messages_fixed = [
    {"role": "system", "content": "你是一个有用的助手，可以查询天气。"},
    {"role": "user", "content": "北京今天天气怎么样？"}
]

# 步骤一：LLM 返回工具调用
response = client.chat.completions.create(
    model=MODEL,
    messages=messages_fixed,
    tools=tools_fixed,
    tool_choice="auto"
)

assistant_message = response.choices[0].message
tool_call = assistant_message.tool_calls[0]

print(f"🔧 LLM 决定调用工具：{tool_call.function.name}")
print(f"🆔 LLM 返回的 tool_call_id：{tool_call.id}")

# 步骤二：执行工具
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
result = TOOL_REGISTRY_FIXED[func_name](**func_args)

print(f"📋 工具执行结果：{result}")

# 步骤三：✅ 正确的消息拼接（使用 LLM 返回的 ID）
messages_fixed.append(assistant_message.model_dump())
messages_fixed.append({
    "role": "tool",
    "tool_call_id": tool_call.id,  # ← 正确：使用 LLM 返回的 ID
    "content": result
})

print(f"\n✅ 修复版本：使用 LLM 返回的 ID '{tool_call.id}'")

# 步骤四：再次调用 LLM
final_response = client.chat.completions.create(
    model=MODEL,
    messages=messages_fixed,
    tools=tools_fixed
)

print(f"🤖 LLM 回答：")
print(final_response.choices[0].message.content)

3.4. 效果验证与对比

消息拼接方式对比：错误版本 vs 修复版本

3.5. 补充说明：消息顺序错误

除了 tool_call_id 不匹配，另一个常见错误是消息顺序错误。==必须先追加 assistant 的工具调用消息，再追加 tool 的结果消息。==如果顺序颠倒，API 会直接报错：

# ❌ 错误示例：消息顺序错误
messages_wrong_order = [
    {"role": "user", "content": "北京今天天气怎么样？"}
]

# 错误：先追加 tool 消息
messages_wrong_order.append({
    "role": "tool",
    "tool_call_id": tool_call.id,
    "content": result
})

# 错误：后追加 assistant 消息
messages_wrong_order.append(assistant_message.model_dump())

# 尝试调用 LLM（会报错）
try:
    response = client.chat.completions.create(
        model=MODEL,
        messages=messages_wrong_order,
        tools=tools_fixed
    )
except Exception as e:
    print(f"❌ API 报错：{e}")
    # 预期错误信息："tool message must follow assistant message with tool_calls"

# ----------------------------------------------------
# ✅ 正确示例：先存 assistant 消息，再存 tool 执行结果
# ----------------------------------------------------
messages_correct = [
    {"role": "system", "content": "你是一个有用的助手，可以查询天气。"},
    {"role": "user", "content": "北京今天天气怎么样？"}
]

try:
    # 第一次调用 LLM
    response = client.chat.completions.create(
        model=MODEL,
        messages=messages_correct,
        tools=tools_fixed,
        tool_choice="auto"
    )

    # 拿到模型返回的原始消息对象
    assistant_message = response.choices[0].message
    tool_call = assistant_message.tool_calls[0]
    
    print(f"🔧 LLM 决定调用工具：{tool_call.function.name}")
    print(f"🆔 拿到 LLM 分配的订单号 ID：{tool_call.id}")

    # ===== 执行本地真实工具 =====
    func_name = tool_call.function.name
    func_args = json.loads(tool_call.function.arguments)
    result = TOOL_REGISTRY_FIXED[func_name](**func_args)
    print(f"📋 工具执行结果：{result}")

    # ===== 拼装历史记录送还给 LLM =====
    # 步骤一：✅ 必须【先】把模型刚才那条带 tool_calls 的消息存入上下文
    # (如果不存，模型会忘记自己曾经下过单)
    messages_correct.append(assistant_message)

    # 步骤二：✅ 然后紧接着存入你的本地执行结果
    # 并且使用严格匹配的订单号 tool_call.id
    messages_correct.append({
        "role": "tool",
        "tool_call_id": tool_call.id,  # ← 取自上面提取的 ID
        "name": func_name,             # （可选但推荐）带上当前执行的函数名
        "content": str(result)         # 必须转化为字符串
    })

    print(f"\n✅ 成功拼接上下文，准备发起第二次回答...")

    # 第二次调用 LLM
    final_response = client.chat.completions.create(
        model=MODEL,
        messages=messages_correct,
        tools=tools_fixed
    )

    print(f"\n🤖 LLM 最终回答：")
    print(final_response.choices[0].message.content)

except Exception as e:
    print(f"❌ 发生了意料之外的错误：{e}")

🔥 踩坑预警：在并行调用多个工具时，所有工具的结果必须在同一轮回传。不能先回传一个工具的结果、调用 LLM、再回传另一个工具的结果。正确的做法是：遍历所有 tool_calls，执行所有工具，将所有结果追加到消息历史后，再调用 LLM。

4. 故障类型四：执行层错误

前面三种故障都发生在"LLM 决策"和"消息传递"环节，而执行层错误发生在"工具真正执行"的环节。表现为：工具执行超时、外部 API 返回异常、返回空值等，导致整个流程中断或 LLM 收到错误的结果。执行层错误的危害最大——如果不做异常处理，一个工具的失败会导致整个 Agent 崩溃。

4.1. 问题复现：异常未捕获导致流程中断

# ❌ 错误示例：定义一个会抛异常的工具
def buggy_get_weather(city: str) -> str:
    """模拟外部 API 调用失败"""
    # 模拟网络超时或 API 错误
    raise Exception("API connection timeout: Unable to reach weather service")

BUGGY_TOOL_REGISTRY = {
    "get_weather": buggy_get_weather,
}

# 测试调用（会崩溃）
messages_buggy = [
    {"role": "system", "content": "你是一个有用的助手，可以查询天气。"},
    {"role": "user", "content": "北京今天天气怎么样？"}
]

response = client.chat.completions.create(
    model=MODEL,
    messages=messages_buggy,
    tools=tools_fixed,
    tool_choice="auto"
)

assistant_message = response.choices[0].message

if assistant_message.tool_calls:
    tool_call = assistant_message.tool_calls[0]
    func_name = tool_call.function.name
    func_args = json.loads(tool_call.function.arguments)

    print(f"🔧 LLM 决定调用工具：{func_name}")
    print(f"📦 参数：{func_args}")

    # 尝试执行工具（会抛异常）
    try:
        result = BUGGY_TOOL_REGISTRY[func_name](**func_args)
        print(f"✅ 执行成功：{result}")
    except Exception as e:
        print(f"\n❌ 流程中断：{e}")
        print(f"💡 问题：异常未被捕获，整个对话流程中断，用户看不到任何回答")

🔧 LLM 决定调用工具：get_weather
📦 参数：{'city': '北京'}

❌ 流程中断：API connection timeout: Unable to reach weather service
💡 问题：异常未被捕获，整个对话流程中断，用户看不到任何回答

这是最糟糕的用户体验——用户提出问题后，系统直接崩溃，没有任何友好的错误提示。

4.2. 问题分析：为什么需要统一异常处理

在真实项目中，工具通常会调用外部 API（天气服务、数据库、搜索引擎等），这些调用都可能失败：

• 网络超时
• API 返回 500 错误
• 返回空值或格式错误的数据
• 权限不足或配额用尽

如果不做异常处理，任何一个工具的失败都会导致整个 Agent 崩溃。核心原则：工具执行失败不应中断主流程，而应该将错误信息转换为结构化的 JSON，传回 LLM，让 LLM 生成友好的降级回答。

4.3. 修复方案：实现统一的异常处理包装器

# ✅ 修复方案：统一的异常处理包装器
def safe_execute_tool(func, func_name: str, args: dict) -> str:
    """
    安全执行工具，统一处理异常

    参数：
        func: 要执行的工具函数
        func_name: 工具名称（用于错误信息）
        args: 工具参数

    返回：
        工具执行结果（JSON 字符串）
        如果执行失败，返回包含错误信息的 JSON
    """
    try:
        result = func(**args)

        # 检查结果是否为空
        if not result or result == "null":
            return json.dumps({
                "warning": f"工具 {func_name} 返回了空结果",
                "args": args,
                "hint": "可能是参数不正确或服务暂时不可用"
            }, ensure_ascii=False)

        return result

    except Exception as e:
        # 将异常转换为 JSON 格式的错误信息
        return json.dumps({
            "error": f"执行 {func_name} 时发生错误",
            "message": str(e),
            "type": type(e).__name__,
            "args": args,
            "hint": "请稍后再试，或联系技术支持"
        }, ensure_ascii=False)

# 测试修复后的版本
messages_fixed = [
    {"role": "system", "content": "你是一个有用的助手，可以查询天气。"},
    {"role": "user", "content": "北京今天天气怎么样？"}
]

response = client.chat.completions.create(
    model=MODEL,
    messages=messages_fixed,
    tools=tools_fixed,
    tool_choice="auto"
)

assistant_message = response.choices[0].message

if assistant_message.tool_calls:
    tool_call = assistant_message.tool_calls[0]
    func_name = tool_call.function.name
    func_args = json.loads(tool_call.function.arguments)

    print(f"🔧 LLM 决定调用工具：{func_name}")
    print(f"📦 参数：{func_args}")

    # 使用安全包装器执行工具
    result = safe_execute_tool(
        BUGGY_TOOL_REGISTRY[func_name],
        func_name,
        func_args
    )

    print(f"⚠️  工具执行失败，但已捕获异常")
    print(f"📋 返回给 LLM 的错误信息：{result}")

    # 将结果回传给 LLM
    messages_fixed.append(assistant_message.model_dump())
    messages_fixed.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result
    })

    # 再次调用 LLM
    final_response = client.chat.completions.create(
        model=MODEL,
        messages=messages_fixed,
        tools=tools_fixed
    )

    print(f"\n🤖 LLM 回答：")
    print(final_response.choices[0].message.content)

虽然工具执行失败了，但流程没有中断，LLM 基于错误信息生成了友好的降级回答。这就是优雅降级的核心思想。

4.4. 效果验证与对比

异常处理方式对比：错误版本 vs 修复版本

4.5. 补充说明：超时控制和空值检查

import time
from functools import wraps

# 超时控制装饰器（简化版）
def with_timeout(timeout_sec: float):
    """装饰器：为函数添加超时控制"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                elapsed = time.time() - start

                if elapsed > timeout_sec:
                    return json.dumps({
                        "error": f"工具执行超时（>{timeout_sec}s）",
                        "elapsed": elapsed,
                        "hint": "请稍后再试或联系技术支持"
                    }, ensure_ascii=False)

                return result
            except Exception as e:
                return json.dumps({
                    "error": f"工具执行异常：{str(e)}",
                    "type": type(e).__name__
                }, ensure_ascii=False)
        return wrapper
    return decorator

# 使用超时控制
@with_timeout(timeout_sec=5.0)
def get_weather_with_timeout(city: str) -> str:
    """带超时控制的天气查询工具"""
    # 模拟耗时操作
    time.sleep(2.0)

    # 简化版：直接返回模拟数据
    weather_db = {
        "北京": {"temperature": 33, "condition": "晴", "humidity": 45},
    }
    data = weather_db.get(city, {"temperature": 25, "condition": "未知", "humidity": 50})
    return json.dumps({"city": city, **data}, ensure_ascii=False)

print("测试超时控制：")
result = get_weather_with_timeout("北京")

💡 实践建议：在生产环境中，建议使用 concurrent.futures.TimeoutError 或 signal.alarm() 实现真正的超时控制。上面的简化版只是演示思路，实际项目中需要更健壮的实现。

5. 故障速查表

Function Calling 四大常见陷阱速查表

使用这个速查表的建议流程：

1. 先看症状：根据你观察到的现象（参数错误、KeyError、回答不相关、流程中断），定位到对应的故障类型
2. 再查根因：理解为什么会出现这个问题
3. 快速排查：按照"快速排查方法"列的指引，用最少的代码验证你的猜测
4. 应用方案：参考"解决方案"列，选择合适的修复方式

💡 实践建议：建议将这个速查表打印出来或保存为书签。在实际项目中遇到 Function Calling 问题时，先查表定位故障类型，再针对性地排查和修复，能节省大量调试时间。

八：并行调用和多函数调用

1. 并行调用的工作原理

当用户的请求需要多个相互独立的工具时，LLM 会在一次响应中返回多个 tool_call 对象。我们的代码遍历这个列表，依次执行每个工具，然后将所有结果一起回传给 LLM。

关键点在于：多个工具的执行结果必须全部回传后，LLM 才会生成最终回答——这意味着如果我们串行执行工具，总耗时是所有工具耗时之和；如果并行执行，总耗时接近最慢那个工具的耗时。

2. 串行 vs 并行性能对比实验

import json
import time
import math
import concurrent.futures
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv(override=True)

client = OpenAI(
    api_key=os.environ.get("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"  # DeepSeek API 端点
)

# ==========================================
# 1. 定义本地真实函数 & 注册表
# ==========================================
def get_weather(location):
    print(f"☁️ [执行工具] 开始查询 {location} 的天气...")
    time.sleep(2.0)  # 模拟 2 秒的网络延迟
    print(f"☁️ [执行工具] 查询完成: {location}")
    return json.dumps({"location": location, "weather": "晴转多云", "temp": "25℃"})

def calculate_sqrt(number):
    print(f"🧮 [执行工具] 开始计算 {number} 的平方根...")
    time.sleep(2.0)  # 模拟 2 秒的计算延迟
    result = math.sqrt(float(number))
    print(f"🧮 [执行工具] 计算完成: {number}")
    return json.dumps({"number": number, "sqrt": result})

# 函数注册表
TOOL_REGISTRY = {
    "get_weather": get_weather,
    "calculate_sqrt": calculate_sqrt
}

# ==========================================
# 2. 面向大模型的工具描述 (JSON Schema)
# ==========================================
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的天气状况",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "要查询的城市名称，例如北京"
                    }
                },
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_sqrt",
            "description": "计算一个数字的平方根",
            "parameters": {
                "type": "object",
                "properties": {
                    "number": {
                        "type": "number",
                        "description": "需要计算平方根的数字"
                    }
                },
                "required": ["number"]
            }
        }
    }
]

# ==========================================
# 3. 串行 vs 并行 的底层调度实现
# ==========================================
def execute_tools_serial(tool_calls):
    """串行执行：一个接一个排队执行"""
    results = []
    start_time = time.time()
    
    for tc in tool_calls:
        func_name = tc.function.name
        func_args = json.loads(tc.function.arguments)
        if func_name in TOOL_REGISTRY:
            res = TOOL_REGISTRY[func_name](**func_args)
            results.append({"id": tc.id, "result": res})
            
    end_time = time.time()
    print(f"⏳ 串行执行总耗时: {end_time - start_time:.2f} 秒")
    return results

def execute_tools_parallel(tool_calls):
    """并行执行：开多线程同时干活"""
    results = []
    start_time = time.time()
    
    # 定义单个线程要干的活
    def _run_single_tool(tc):
        func_name = tc.function.name
        func_args = json.loads(tc.function.arguments)
        if func_name in TOOL_REGISTRY:
            res = TOOL_REGISTRY[func_name](**func_args)
            return {"id": tc.id, "result": res}
        return None

    # 使用 Python 原生的线程池实现并发调用
    with concurrent.futures.ThreadPoolExecutor() as executor:
        # executor.map 会自动开多线程执行，并且最后收集结果时仍保持原本的顺序
        results = list(executor.map(_run_single_tool, tool_calls))
        
    end_time = time.time()
    print(f"🚀 并行执行总耗时: {end_time - start_time:.2f} 秒")
    return results

# ==========================================
# 以下是你提供的调用测试代码
# ==========================================
MODEL = "deepseek-chat"  # 替换成你实际可用的模型，比如 "deepseek-chat" 或 "gpt-4o"

print("🤖 发送并行请求给大模型...")
response = client.chat.completions.create(
    model=MODEL,
    messages=[
        {"role": "system", "content": "你是一个有用的助手。"},
        {"role": "user", "content": "帮我查一下北京的天气，同时计算 sqrt(256)"}
    ],
    tools=tools,
    tool_choice="auto",
    temperature=0.7,
    max_tokens=512
)
assistant_msg = response.choices[0].message

if assistant_msg.tool_calls:
    print(f"\n✅ LLM 决定同时调用 {len(assistant_msg.tool_calls)} 个工具：")
    for tc in assistant_msg.tool_calls:
        print(f"  - {tc.function.name} 参数: {tc.function.arguments}")

    print("\n--- 【对比测试开始】 ---")
    
    print("\n[模式 A] 串行执行（慢）：")
    serial_results = execute_tools_serial(assistant_msg.tool_calls)

    print("\n[模式 B] 并行执行（快）：")
    parallel_results = execute_tools_parallel(assistant_msg.tool_calls)
else:
    print("LLM 直接回答，未调用工具")

🤖 发送并行请求给大模型...

✅ LLM 决定同时调用 2 个工具：
  - get_weather 参数: {"location": "北京"}
  - calculate_sqrt 参数: {"number": 256}

--- 【对比测试开始】 ---

[模式 A] 串行执行（慢）：
☁️ [执行工具] 开始查询 北京 的天气...
☁️ [执行工具] 查询完成: 北京
🧮 [执行工具] 开始计算 256 的平方根...
🧮 [执行工具] 计算完成: 256
⏳ 串行执行总耗时: 4.00 秒

[模式 B] 并行执行（快）：
☁️ [执行工具] 开始查询 北京 的天气...
🧮 [执行工具] 开始计算 256 的平方根...
☁️ [执行工具] 查询完成: 北京
🧮 [执行工具] 计算完成: 256
🚀 并行执行总耗时: 2.01 秒

运行后你会看到明显的性能差异：串行执行的总耗时是所有工具耗时之和，而并行执行的总耗时接近最慢那个工具的耗时。在真实项目中，如果一次请求需要调用 5 个各耗时 2 秒的 API，串行需要 10 秒，并行只需要 2 秒——性能差距随工具数量线性放大。

本节课程我们实现的 Function Calling 管线有一个关键限制：它是单轮的。LLM 调用一次工具、获取一次结果、生成一次回答，整个流程就结束了。但现实中的复杂任务往往需要多步推理——例如"先搜索 LangChain 的最新版本号，再搜索该版本的 changelog，最后总结主要变化"。这需要 LLM 在获取第一步结果后，基于结果决定下一步行动，形成一个推理-行动的循环。